package com.alibaba.nacos.core.remote.grpc;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.grpc.auto.Payload;
import com.alibaba.nacos.api.remote.DefaultRequestFuture;
import com.alibaba.nacos.api.remote.RequestCallBack;
import com.alibaba.nacos.api.remote.request.Request;
import com.alibaba.nacos.api.remote.response.Response;
import com.alibaba.nacos.common.remote.client.grpc.GrpcUtils;
import com.alibaba.nacos.core.remote.Connection;
import com.alibaba.nacos.core.remote.ConnectionMeta;
import com.alibaba.nacos.core.remote.RpcAckCallbackSynchronizer;
import io.grpc.netty.shaded.io.netty.channel.Channel;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.UUID;

@Slf4j
public class GrpcConnection
        extends Connection {

    private StreamObserver emitter;
    private Channel channel;

    public GrpcConnection(ConnectionMeta connectionMeta,
                          StreamObserver emitter,
                          Channel channel) {
        super(connectionMeta);
        this.emitter = emitter;
        this.channel = channel;
    }

    @Override
    public boolean isConnected() {
        return channel != null && channel.isOpen() && channel.isActive();
    }

    @Override
    public void asyncRequest(Request request,
                             RequestCallBack requestCallBack) throws NacosException {
        sendRequestInner(request, requestCallBack);
    }

    private DefaultRequestFuture sendRequestInner(Request request,
                                                  RequestCallBack requestCallBack) throws NacosException {
        String traceId = UUID.randomUUID().toString();
        request.setRequestId(traceId);

        String connectionId = getMetaInfo().getConnectionId();

        DefaultRequestFuture requestFuture = new DefaultRequestFuture().setConnectionId(connectionId)
                .setRequestId(traceId)
                .setRequestCallBack(requestCallBack)
                .setTimeoutInnerTrigger(() -> RpcAckCallbackSynchronizer.clearFuture(connectionId, traceId));
        RpcAckCallbackSynchronizer.syncCallback(connectionId, traceId, requestFuture);
        sendRequestNoAck(request);
        return requestFuture;
    }

    @SneakyThrows
    @Override
    public Response request(Request request,
                            long timeoutLills) throws NacosException {
        DefaultRequestFuture requestFuture = sendRequestInner(request, null);
        return requestFuture.get();
    }

    private void sendRequestNoAck(Request request) {
        sendQueueBlockCheck();
        Payload payload = GrpcUtils.convert(request);
        emitter.onNext(payload);
        // 双向流式 不能completed
    }

    private void sendQueueBlockCheck() {
        if (emitter instanceof ServerCallStreamObserver streamObserver) {
            boolean ready = streamObserver.isReady();
            if (ready) {
                getMetaInfo().clearPushQueueBlockTimes();
            } else {

            }
        }
    }
}
